package com.chukun.flink.dataset.operator;

import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.MapPartitionFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.util.Collector;

import java.util.Iterator;

/**
 * @author chukun
 * @version 1.0.0
 * @description mapPartition 操作符的基本操作
 * @createTime 2022年05月28日 20:35:00
 */
public class MapPartitionOperator {

    public static void main(String[] args) throws Exception {

        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(2);

        DataSource<Long> dataSource = env.generateSequence(1, 5);

        DataSet<String> mappedData = dataSource.mapPartition(new MapPartitionFunction<Long,String>() {

            @Override
            public void mapPartition(Iterable<Long> iterable, Collector<String> collector) throws Exception {
                long count = 0;
                String result = "";
                Iterator<Long> iterator = iterable.iterator();
                while (iterator.hasNext()) {
                    Long elem = iterator.next();
                    count++;
                    result = StringUtils.join(result, ",", elem);
                }
                collector.collect("分区中迭代器内的元素数量: "+count +" 元素列表: "+result);
            }
        });

        mappedData.print("MapPartitionFunction");

        env.execute("MapPartitionOperator");
    }
}
